package edu.ucla.cs.rpc.multicast.commhistory;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Set;

import edu.ucla.cs.rpc.multicast.Dest;
import edu.ucla.cs.rpc.multicast.Sender;
import edu.ucla.cs.rpc.multicast.handlers.MessageHandler;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageReceiver;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageSender;
import edu.ucla.cs.rpc.multicast.network.message.Message;
import edu.ucla.cs.rpc.multicast.util.IdentifierComparator;

/**
 * This class implements the causal history variant of the communications
 * history scheme discussed in the DeFago paper. Each CausalProcess is both a
 * sender and a destination.
 * 
 * @author Philip Russell Chase Covello
 * @see Total Order Broadcast and Multicast Algorithms: Taxonomy and Survery,
 *      Defago et al
 * @see Section 4.4, pages 16 - 18
 * 
 */
public class CausalProcess implements Dest, Sender {

	private MessageHandler handler;

	/* The set of received messages, as stated in the paper */
	private Set<Message> received = null;

	/* The set of messages that have been delivered */
	private Set<Message> delivered = null;

	/*
	 * The logical clocks for each process in the system, according to this
	 * process. If we view this as a logical array, each cell is initially zero,
	 * and is updated upon receiving a time-stamped message
	 */
	private Hashtable<Long, Long> logicalClocks = null;

	/*
	 * The communications agent that receieves messages from other
	 * CausalProcesses
	 */
	private MulticastMessageReceiver receiever = null;

	/* The communications agent that sends messages to other CausalProcesses */
	private MulticastMessageSender sender = null;

	/**
	 * Primary constructor.
	 * 
	 * @param manager
	 *            The multicast communications agent that manages group routing
	 *            FROM a group of CausalProcesses, and TO a group of
	 *            CausalProcesses.
	 * @param handler
	 *            The upper-level process that receives messages.
	 * @throws IOException
	 */
	public CausalProcess(SocketAddress manager, MessageHandler handler)
			throws IOException {

		this.handler = handler;
		/*
		 * We'll be receiving from a GROUP of CausalProcesses, so initialize the
		 * receiver to link with the MultiCastManager
		 */
		this.receiever = new MulticastMessageReceiver(manager,
				new CausalProcessMessageHandler());
		/*
		 * We'll be sending to a GROUP of CausalProcesses, so initialize the
		 * receiver to link with the MultiCastManager
		 */
		this.sender = new MulticastMessageSender(manager);

		/* Both RECEIVED and DELIVERED are sets of unique messages */
		this.received = new HashSet<Message>();
		this.delivered = new HashSet<Message>();

		/*
		 * Initial the logical clocks. It it better to view this as a logical
		 * array, mapping an index (the process id) to a value (the last known
		 * logical clock value of that process id)
		 */
		Set<Long> processSet = this.receiever.getSenders();
		this.logicalClocks = new Hashtable<Long, Long>(processSet.size() + 1,
				(float) 1.0);
		/*
		 * For each process in the set, create a new data pair of its ID and the
		 * number 0. Analogous to the third line of the initialization phase of
		 * the algorithm, page 17 of the DeFago paper
		 */
		for (Long elem : processSet) {
			this.logicalClocks.put(elem, (long) 0);
		}
	}

	/**
	 * Primary functionality of a CausalProcess. Implements the algorithm
	 * starting on page 17 of the DeFago paper, starting at the event "when
	 * receive (m, ts(m))". It is the event handler for receiving time-stamped
	 * messages.
	 * 
	 * @author Philip Russell Chase Covello
	 * 
	 */
	private class CausalProcessMessageHandler implements MessageHandler {
		public void receive(Message message) {

			/*
			 * Get the id of this process; this is the variable P in the
			 * algorithm
			 */
			long index = receiever.getNodeID();
			long clockvalue = logicalClocks.get(index);
			/*
			 * Update this process's clock value to be the maximum of its
			 * current clock value and the timestamp of the incoming message
			 */
			logicalClocks.put(index, Math.max(message.getTimestamp(),
					clockvalue));

			long messageSenderIndex = message.getSenderID();

			/* Update the sender's logical clock from P's perspective */
			logicalClocks.put(messageSenderIndex, message.getTimestamp());

			/* Add the message to the set of received messages */
			received.add(message);

			/* Get the set difference of RECEIVED and DELIVERED */
			Set<Message> difference = new HashSet<Message>(received);
			difference.removeAll(delivered);

			/* Construct the set of DELIVERABLE messages */
			List<Message> deliverable = new ArrayList<Message>();

			/* Get the minimum clock value */
			long minClockTime = Collections.min(logicalClocks.values());
			for (Message potentialMessage : difference) {
				/*
				 * For each message in the set difference, if the message's
				 * timestamp is less than the minimum clock value of ALL
				 * processes from process P's perspective, then add the message
				 * to DELIVERABLE
				 */
				if (potentialMessage.getTimestamp() < Math.min(potentialMessage
						.getTimestamp(), minClockTime)) {
					deliverable.add(potentialMessage);
				}

			}

			/*
			 * Sort the message in DELIVERABLE according to (ts(m), sm(m))
			 * order. Ie By timestamp, by sender ID for ties
			 */
			Collections.sort(deliverable, new IdentifierComparator());

			/* Deliver each message in deliverable to the upper level */
			for (Message deliverMessage : deliverable) {
				handler.receive(deliverMessage);
			}

			/* Add each delivered message to the list of such messages */
			delivered.addAll(deliverable);
		}
	}

	/**
	 * Shuts down both the message sending and message receiving functionality.
	 */
	public void shutdown() {
		receiever.shutdown();
		sender.shutdown();
	}

	/**
	 * Implemenation of the procedure TO-multicast(m), page 17 of the DeFago
	 * paper.
	 * 
	 * @param message
	 *            The message to broudcast.
	 */
	public void totalOrderBroadcast(Message message) throws IOException {
		/* Get the most recent clock value for this process */
		long timestamp = this.logicalClocks.get(this.receiever.getNodeID());
		timestamp++;
		message.setTimestamp(timestamp);

		/* Send the message */
		this.sender.send(message);

		/* Update this process's logical clock */
		this.logicalClocks.put(this.receiever.getNodeID(), timestamp);
	}

	/**
	 * @param handler
	 */
	public void setMessageHandler(MessageHandler handler) {
		this.handler = handler;
	}
}
